package com.sf.shiva.oms.jstorm.bolt;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sf.shiva.oms.jstorm.common.enumtype.FieldKeyEnum;
import com.sf.shiva.oms.jstorm.entity.demo.PackageStatusEntity;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

/**
 * 
 * 描述：计算bolt
 *
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2018年4月2日      01369626         Create
 * ****************************************************************************
 * </pre>
 * @author 01369626
 * @since 1.0
 */
public class CalcBolt implements IRichBolt{

    /**  */
    private static final long serialVersionUID = 1L;
    
    private static final Logger logger = LoggerFactory.getLogger(CalcBolt.class);
    
    private transient OutputCollector collector;
    
    /*
     * prepare在bolt被nimbus分发到各个节点后，worker初始化task时调用。因此，可以把一些组件的初始化放在这里，避免因为组件在nimbus分发到supervisor时由于
     * 序列化执行失败，导致任务启动失败；这里prepare方法提供给bolt一个OutputCollector用来发射tuple
     */
    @Override
    public void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }
    
    /*
     * bolt执行处理逻辑的方法，一般建议在这里捕获所有异常进行处理，避免因异常导致task挂掉
     */
    @Override
    public void execute(Tuple input) {
        try {
            PackageStatusEntity entity = (PackageStatusEntity)input.getValueByField(FieldKeyEnum.CUSTOM_FIELD_1.getKey());
            //calc();执行相应的处理逻辑
            collector.emit(input, new Values(entity));//消息发送给下游,调用emit(anchor, tuple)方法对消息进行锚定，才能触发jstorm的ack机制
            collector.ack(input);//当前节点处理成功，发送ack确认
        }catch (Exception e) {
            logger.error("CalcBolt execute error.", e);
            //发生异常时，需要调用fail方法让jstorm执行重新发送tuple，避免因处理失败导致数据丢失
            collector.fail(input);
        }
    }

    @Override
    public void cleanup() {
      //在拓扑停止时执行，可以用来实现数据库连接等资源释放
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(FieldKeyEnum.CUSTOM_FIELD_2.getKey()));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

}
